package cn.hfuu.spark;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * @author oldliu
 * @since 1.0
 */
public class WordCount2 {
    public static void main(String[] args) {
        // 创建SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("SparkApp")
                .master("local")
                .getOrCreate();

        // 创建JavaSparkContext
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        JavaRDD<String> lines = jsc.textFile("hdfs://hadoopmaster:9000/heda2022");

        //?String n = lines.max((x, y) -> x.length());
        //?System.out.println("最大的是 is :" + n);

        JavaPairRDD<Integer, String> map =
                lines.flatMap(x -> Arrays.asList(x.split("(\\s+|\\.|,|!|:|;|\")")).iterator()).mapToPair(word -> new Tuple2<>(word, 1)).reduceByKey((a, b) -> a + b).mapToPair(k -> new Tuple2<Integer, String>(k._2, k._1));
        //降序

        List<Tuple2<String, Integer>> res = map.sortByKey(false).mapToPair(k -> new Tuple2<String, Integer>(k._2,
                k._1)).collect();

        System.out.println("排序结果如下");
        res.forEach(k -> System.out.println(k._1 + "===>" + k._2));
    }
}
